package com.sweet.logging.nio;

import com.sweet.logging.appender.RemoteAppender;
import com.sweet.logging.logger.LoggingEvent;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * 向日志采集服务器推送日志
 *
 *    考虑到日志吞吐量比较高，这里用长连接。如果不幸掉线中断
 *    则进行不断的重试工作
 *
 * @author xiexiaofeng
 * @version 1.0     2017年09月08日  14点33分
 */
public class SweetRemoteClient
{

    private static final byte[] spaceBytes = " ".getBytes();

    private Selector       selector = null;
    private SocketChannel  client   = null;
    private RemoteAppender appender = null;
    private Charset        charset  = Charset.forName("UTF-8");
    private ByteBuffer     buff     = ByteBuffer.allocate(2048); // 输出缓冲
    private String server;
    private int    port;

    public SweetRemoteClient(String server, int port, RemoteAppender appender) throws IOException
    {
        synchronized (SweetRemoteClient.class)
        {
            this.server = server;
            this.port = port;
            this.appender = appender;
            connect();
        }
    }

    /**
     * 建立连接
     * @throws IOException 连接失败
     */
    private synchronized void connect() throws IOException
    {
        this.client = SocketChannel.open(new InetSocketAddress(server, port));
        this.client.configureBlocking(false);
        this.selector = Selector.open();
        this.client.register(selector, SelectionKey.OP_WRITE);
    }

    /**
     * 如果发生掉线，比如服务器暂时不在线等，客户端每三秒会重连一次
     * @return 重连成功
     */
    private boolean reConnection()
    {
        synchronized (this)
        {
            try
            {
                connect();
            } catch (IOException e)
            {
                e.printStackTrace();
                return false;
            }
        }
        return true;
    }

    public void start()
    {
        new Worker().start();
    }

    public void close()
    {
        try
        {
            selector.close();
            client.close();
        } catch (IOException e)
        {
            e.printStackTrace();
        }

    }

    private volatile boolean connecting = true;

    private class Worker extends Thread
    {
        public void run() {
            //轮询
            while(true)
            {
                try
                {
                    if (!connecting)
                    {
                        for (;!reConnection();)
                            TimeUnit.SECONDS.sleep(1); // 等一会在重连
                    }
                    int readyChannels = selector.select();
                    if(readyChannels == 0) continue;
                    Set<SelectionKey> selectedKeys = selector.selectedKeys();
                    Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
                    while(keyIterator.hasNext())
                    {
                        SelectionKey key = keyIterator.next();
                        keyIterator.remove();
                        process(key);
                    }
                }
                catch (IOException io)
                {
                    try
                    {
                        TimeUnit.SECONDS.sleep(1); // 等一会再试试
                    } catch (InterruptedException e)
                    {
                        e.printStackTrace();
                    }
                } catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }
        }

        private int failCount = 0;

        private void process(SelectionKey key) throws IOException
        {
            try
            {
                if (key.isWritable())
                {
                    ByteBuffer byteBuffer = serialize(buff, appender.take());
                    if (byteBuffer != null)
                    {
                        byteBuffer.flip();
                        try
                        {
                            client.write(byteBuffer);
                            failCount = 0;
                        } catch (Exception e) {
                            // 写入失败
                            if (failCount >= 3)
                                connecting = false;
                            failCount++;
                        }
                        byteBuffer.compact();
                        key.interestOps(SelectionKey.OP_WRITE);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        /**
         * 序列化
         * @param buffer 缓冲
         * @param event 日志信息
         */
        private ByteBuffer serialize(ByteBuffer buffer, LoggingEvent event)
        {
            if (event == null)
                return null;
            buffer.put(String.valueOf(event.getTimeStamp()).getBytes());
            buffer.put(spaceBytes);
            buffer.put(event.getLevel().getName().getBytes(charset));
            buffer.put(spaceBytes);
            buffer.put(event.getName().getBytes(charset));
            buffer.put(spaceBytes);
            int need = event.getMessage().length() * 2 + 2;
            if (need > buffer.remaining())
            {
                buffer = ByteBuffer.allocate(buffer.position() + need).put(buffer);
            }
            buffer.put((event.getMessage() + Character.MIN_VALUE).getBytes(charset));
            return buffer; // 如果给定的buffer不够用，考虑换个大的
        }


    }

}



